package com.lagou

import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

object WorkTwo {
  def main(args: Array[String]): Unit = {
    // 1、创建SparkContext
    Logger.getLogger("org").setLevel(Level.WARN)
    val spark = SparkSession.builder()
      .appName(s"${this.getClass.getCanonicalName}")
      .master("local[*]")
      .getOrCreate()
    val sc = spark.sparkContext
    val logRDD: RDD[(String, String, String)] = sc.textFile("data/source2/cdn.txt").map {
      line =>
        val fields = line.split("\\s+")
        val ip = fields(0)
        val hour: String = fields(3).split("/").last.split(":")(1)
        val website = fields(6)
        (ip, hour, website)
    }
    val df: DataFrame = spark.createDataFrame(logRDD).toDF("ip", "hourOfDay", "website")
    df.createOrReplaceTempView("log")

    // 计算独立IP数
    spark.sql(
      """
        |select count(distinct(ip)) as count from log
        |""".stripMargin).show()

    // 统计每个视频独立IP数
    spark.sql(
      """
        |select website, count(distinct(ip)) as ipCount from
        | ( select ip,website from log where website rlike '.*\.mp4$' ) t
        | group by website order by count(distinct(ip)) desc
        |""".stripMargin).show(10, truncate = false)

    // 统计一天中每个小时的流量
    spark.sql(
      """
        |select hourOfDay, count(*) as count from log
        |group by hourOfDay order by hourOfDay
        |""".stripMargin).show()

    spark.close()
  }
}
